package user.service.Impl;

import com.alibaba.fastjson.JSON;
import com.spring4all.spring.boot.starter.hbase.api.HbaseTemplate;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.PrefixFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import user.contstant.Constants;
import user.mapper.FeedBackRowMapper;
import user.service.IFeedBackService;
import user.util.RowKeyGenUtil;
import user.vo.FeedBack;
import user.vo.Response;

import java.util.List;

/**
 * 评论共能实现
 */
@Slf4j
@Service
public class FeedBackServiceImpl implements IFeedBackService {

    /**HBase 客户端*/
    private final HbaseTemplate hbaseTemplate;

    @Autowired
    public FeedBackServiceImpl(HbaseTemplate hbaseTemplate) {
        this.hbaseTemplate = hbaseTemplate;
    }

    //创建评论
    @Override
    public Response createFeedBack(FeedBack feedBack) {

        /** 判断feedback是否合法*/
        if(!feedBack.validate()){
            log.error("feedback error:{}", JSON.toJSONString(feedBack));
            //返回错误响应
            return Response.failure("feedback error");
        }

        /** 写入HBase*/
        Put put = new Put(Bytes.toBytes(RowKeyGenUtil.genFeedBackRowKey(feedBack)));
        put.addColumn(Bytes.toBytes(Constants.FeedbackTable.FAMILY_I),
                      Bytes.toBytes(Constants.FeedbackTable.USER_ID),
                      Bytes.toBytes(feedBack.getUserId())
                );
        put.addColumn(Bytes.toBytes(Constants.FeedbackTable.FAMILY_I),
                Bytes.toBytes(Constants.FeedbackTable.TEMPLATE_ID),
                Bytes.toBytes(feedBack.getTemplateId())
        );
        put.addColumn(Bytes.toBytes(Constants.FeedbackTable.FAMILY_I),
                Bytes.toBytes(Constants.FeedbackTable.TYPE),
                Bytes.toBytes(feedBack.getType())
        );
        put.addColumn(Bytes.toBytes(Constants.FeedbackTable.FAMILY_I),
                Bytes.toBytes(Constants.FeedbackTable.COMMENT),
                Bytes.toBytes(feedBack.getComment())
        );

        hbaseTemplate.saveOrUpdate(Constants.FeedbackTable.TABLE_NAME,put);

        //成功响应
        return Response.success();
    }

    //获取用户评论
    @Override
    public Response getFeedBack(Long userId) {

        //获取反转后的userId字节数组
        byte[] resverUserId = new StringBuilder(String.valueOf(userId)).reverse().toString().getBytes();

        /** 定义扫描器，扫描HBase*/
        Scan scan = new Scan();
        //前缀过滤器扫描
        scan.setFilter(new PrefixFilter(resverUserId));
        //序列化成java对象
        List<FeedBack> feedBacks = hbaseTemplate.find(Constants.FeedbackTable.TABLE_NAME,scan,new FeedBackRowMapper());
        return new Response(feedBacks);
    }
}
